package utils

import (
	"fmt"
	"io"
	"sync"
	"sync/atomic"
)

type WriteQueueTask struct {
	terminated int8
	closewg    sync.WaitGroup
	task       *MultiGroupTask
	lst        GSyncMapGroup
	memPool    *MemBlockPool
}

func NewWriteQueueTask(blocksize uint16, maxwork int) *WriteQueueTask {
	rval := &WriteQueueTask{}
	rval.task = NewMultiGroupTask()
	rval.task.ConfigWorkType(2)
	rval.task.ConfigMinWork(1)
	rval.task.ConfigMaxWork(int32(maxwork))
	rval.memPool = NewMemBlockPool(blocksize)
	return rval
}

func (this *WriteQueueTask) Start() {
	this.terminated = 0
	this.task.Start()
}

func (this *WriteQueueTask) Stop() {
	this.terminated = 1
	this.closewg.Wait()
	this.task.Stop()
}

func (this *WriteQueueTask) MemPoolStatus() string {
	n := this.memPool.GetBlockAliveNum()
	return fmt.Sprintf("block:%d*%d=%s", n, this.memPool.block_size, HumanFilesize(int64(n)*int64(this.memPool.block_size)))
}

func (this *WriteQueueTask) TaskStatus() string {
	return this.task.StatusSimpleString()
}

/*
queue_write_func 不可以异步写入, 因为是内存块是需要回收的
*/
func (this *WriteQueueTask) NewSession(id string, maxsize int, queue_write_func func(p []byte) (n int, err error)) io.Writer {
	queue := NewMemBlockSyncListEx(this.memPool)
	queue.ConfigMaxSize(maxsize)
	rval := &writeQueueSession{id: id, queue: queue, owner: this, queue_write_func: queue_write_func}
	this.lst.Set(id, rval)
	return rval
}

func (this *WriteQueueTask) RemoveSession(id string) {
	obj, _ := this.lst.RemoveAndGet(id)
	if obj != nil {
		sess := obj.(*writeQueueSession)
		sess.owner = nil
		sess.queue_write_func = nil
	}
}

func (this *WriteQueueTask) SendOnce(args ...interface{}) {
	if len(args) == 0 {
		return
	}

	sess, _ := args[0].(*writeQueueSession)
	if sess == nil {
		return
	}
	atomic.StoreInt32(&sess.writeflag, 1)
	defer atomic.StoreInt32(&sess.writeflag, 0) // 直接置为0

	writer_func := sess.queue_write_func
	sendQueue := sess.queue
	if sendQueue == nil || writer_func == nil {
		return
	}

	if sendQueue.Size() == 0 {
		return
	}

	this.closewg.Add(1)
	defer this.closewg.Done()

	for {
		block := sendQueue.PopMemBlock()
		if block == nil {
			if atomic.CompareAndSwapInt32(&sess.writeflag, 1, 0) {
				break
			}
			continue
		}

		buf := block.GetBytes()
		if this.terminated == 0 {
			_, err := writer_func(buf)
			block.Release2Pool()
			if err != nil {
				break
			}
		} else {
			block.Release2Pool()
		}
	}

}

type writeQueueSession struct {
	writeflag        int32
	id               string
	queue_write_func func(p []byte) (n int, err error)
	owner            *WriteQueueTask
	queue            *MemBlockSyncList
}

func (this *writeQueueSession) Write(buf []byte) (n int, err error) {
	owner := this.owner
	if owner == nil {
		return -1, fmt.Errorf("writer is closed!")
	}
	n, err = this.queue.Write(buf)
	flag := atomic.LoadInt32(&this.writeflag)
	if flag != 1 {
		owner.task.PostTaskFuncArgs(this.id, owner.SendOnce, this)
	}
	return
}
